package com.wasp.collector.service.akka;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.Lists;
import com.wasp.collector.data.RequestContext;
import com.wasp.collector.data.TimeInterval;
import com.wasp.collector.service.ICollector;
import com.wasp.collector.utils.EmailUtil;

import akka.actor.ActorRef;

public class ExceptionMonitor extends Thread {
	ICollector collect = null;
	boolean isRunning = true;
	private Object objLock = new Object();
	private static final Logger log = LoggerFactory.getLogger(ExceptionMonitor.class);


	public ExceptionMonitor(ICollector collect) {
		this.collect = collect;
		this.setName(String.format("ExceptionMonitor:%s", ""));
	}

	@Override
	public void run() {
		while (isRunning) {
			Map<Date,RequestContext> errorStocksOccurs = collect.getErrors();
			synchronized (objLock) {
				while (true) {
					if (errorStocksOccurs.size() == 0) {
						try {
							objLock.wait(3000);
						} catch (InterruptedException e) {
							log.error("",e);
						}
					}
					if (collect.getErrors().size() > 0) {
						break;
					}
				}
			}
			try {
				checkCollect();
			} catch (Exception e) {
				log.error("",e);
			}
		}
	}

	public void stopMonitor() {
		this.isRunning = false;
	}

	private void slowDown(ICollector collect, int interval, boolean isStop) {
		List<ActorRef> listActors = Lists.newArrayList(collect.getRegisterActors().values());
		for (ActorRef a : listActors) {
			if (a != null && !a.isTerminated()) {
				// 异常停止采集时，发送邮件
				if (isStop) {
					String emailTitle = "异常提醒";
					String address = "大哥";
					String subject = "数据采集异常";
					String content = "数据采集异常频繁，已停止采集！停止采集时间" + new Date();
					try {
						EmailUtil.sendEmail(emailTitle, address, subject, content);
					} catch (Exception e) {
						log.error("",e);
					}
				}
				TimeInterval in = new TimeInterval();
				in.setInterval(interval);
				in.setStop(isStop);
				a.tell(in, null);
			}
		}
	}

	private void checkCollect() {
		Map<Date,RequestContext> errorStocksOccurs = collect.getErrors();
		boolean isNeedSlowDown = false;
		boolean isNeedStop = false;
		// 等待间隔时间，单位毫秒
		int interval = 500;
		Date nowDate = new Date();
		long nowTime = nowDate.getTime();
		long beforeTime = nowTime - (60 * 1000);
		int count = errorStocksOccurs.keySet().stream().filter(errorDate->(errorDate.getTime() > beforeTime && errorDate.getTime() < nowTime))
			.collect(Collectors.toList()).size();
		if (count < 10) {
			// 如果errors很少，可以不等待
			interval = 0;
		} else if (count < 20 && count >= 10) {
			// 循环遍历1分钟内错误的个数，如果大于10，则调用slowdown
			interval = 500;
			isNeedSlowDown = true;
		} else if (count >= 20) {
			// 如果大于20，可以结束采集
			interval = 0;
			isNeedSlowDown = true;
			isNeedStop = true;
		}
		if (isNeedStop) {
			stopMonitor();
		}
		if (isNeedSlowDown) {
			slowDown(collect, interval, isNeedStop);
		}
	}
}
